package com.atguigu.flink.datastream.transform;

import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.nio.charset.StandardCharsets;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 14:06
 *
 *  富函数: 相较于普通函数， 富函数拥有更强大的功能。
 *     RichFunction
 *        AbstractRichFunction
 *           RichMapFunction
 *           RichFilterFunction
 *           RichFlatmapFunction
 *           RichReduceFunction
 *           .......
 *
 * 富函数的功能:
 *
 *    1. 拥有算子生命周期方法
 *        open() : 算子被创建时，执行一次。
 *        close(): 算子被销毁时，执行一次。
 *
 *    2. 获取到运行时上下文 RuntimeContext
 *          可以进行状态编程
 *              getState()
 *              getListState()
 *              getMapState()
 *              getReducingState()
 *              getAggregatingState()
 *
 *          获取更多其他的信息.
 */
public class Flink05_RichFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取socket文本数据，将数据封装成Event，打印输出
        //DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);

        //读取文本数据
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                new TextLineInputFormat(StandardCharsets.UTF_8.name())
                ,
                new Path("input/clicks.txt")
        ).build();
        DataStreamSource<String> ds = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");

        ds.map(
                new RichMapFunction<String, Event>() {
                    /**
                     * 算子在创建的时候执行一次
                     */
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        System.out.println("Flink05_RichFunction.open");
                    }

                    @Override
                    public Event map(String line) throws Exception {
                        System.out.println("Flink05_RichFunction.map");

                        //获取更多的信息
                        //getRuntimeContext().getXXXX

                        String[] fields = line.split(",");
                        return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                    }

                    /**
                     * 算子在结束的时候执行一次.
                     */
                    @Override
                    public void close() throws Exception {
                        System.out.println("Flink05_RichFunction.close");
                    }
                }
        ).setParallelism(2)
                .print();


        env.execute();

    }
}
